package com.sailors.common.task.sdk.core.handle.worker;

import com.alibaba.ttl.TransmittableThreadLocal;
import com.sailors.common.task.sdk.core.model.FileTaskFailedDto;
import com.sailors.common.task.sdk.core.model.RefreshProcessDto;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;

/**
 * Task container
 *
 * @description:
 * @author: cyj
 * @date 2022 -11-25 14:54:28
 */
@Slf4j
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class TaskContainer {


    protected static final BlockingQueue<FileTaskFailedDto> FAILED_EXECUTE_QUEUE = new ArrayBlockingQueue(512);

    protected static final BlockingQueue<RefreshProcessDto> REFRESH_PROCESS_EXECUTE_QUEUE = new ArrayBlockingQueue(8192);

    protected static final ConcurrentHashMap<String, Semaphore> TASK_SEMAPHORE = new ConcurrentHashMap<>();

    protected static  final ThreadLocal<Long> TASK_ID = new TransmittableThreadLocal<Long>();

    /**
     * Aquire 获取信号量
     *
     * @param bussiness bussiness
     * @param limit     limit
     */
    public static void aquire(String bussiness, Integer limit) {
        if (!TASK_SEMAPHORE.containsKey(bussiness)) {
            Semaphore semaphore = new Semaphore(limit);
            TASK_SEMAPHORE.put(bussiness, semaphore);
        }
        if (!TASK_SEMAPHORE.get(bussiness).tryAcquire()) {
            throw new RuntimeException("此业务进行了并行任务限制，当前系统节点有：" + limit + " 个此类型任务正在进行，请稍后重试");
        }
    }

    /**
     * Release 释放信号量
     *
     * @param bussiness bussiness
     */
    public static void release(String bussiness) {
        if (TASK_SEMAPHORE.containsKey(bussiness)) {
            TASK_SEMAPHORE.get(bussiness).release();
        }
    }

    /**
     * Gets task id 获取执行中得任务id
     *
     * @return the task id
     */
    public static Long getTaskId() {
        return TASK_ID.get();
    }

    /**
     * Sets task id 设置taskId
     *
     * @param taskId task id
     */
    public static void setTaskId(Long taskId) {
        TASK_ID.set(taskId);
    }

    /**
     * Remove task id 移除执行中taskId
     */
    public static void removeTaskId() {
        TASK_ID.remove();
    }

    /**
     * Refresh process 通知sdk任务进度刷新
     *
     * @param refreshProcessDto refresh process dto
     */
    public static void refreshProcess(RefreshProcessDto refreshProcessDto) {
        try {
            REFRESH_PROCESS_EXECUTE_QUEUE.put(refreshProcessDto);
        } catch (InterruptedException e) {
            log.warn("TaskContainer.refreshProcess failed:" + e.getMessage(), e);
        }
    }

    /**
     * Faild 通知sdk任务失败
     *
     * @param fileTaskFailedDto file task failed dto
     */
    public static void faild(FileTaskFailedDto fileTaskFailedDto) {
        try {
            FAILED_EXECUTE_QUEUE.put(fileTaskFailedDto);
        } catch (InterruptedException e) {
            log.warn("TaskContainer.faild failed:" + e.getMessage(), e);
        }
    }

}
